package com.leilei;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/20 20:11
 */
public class Flink_SQL_3_function {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);
        DataStreamSource<GirlFriend> scFriendList = env.fromCollection(Arrays.asList(new GirlFriend("小丽", "A", 22),
                new GirlFriend("小丽于", "C", 24),
                new GirlFriend("小胡", "D", 23)));
        Table table = tableEnv.fromDataStream(scFriendList, $("name"), $("cup"), $("age"));
        Table tableResult = tableEnv.sqlQuery("select avg(`age`) as avgAge,sum(age) as totalAge from " + table);
        // boolean 更新的数据为true,未更新为false; row(模糊数据行，如果结果能映射到实体，建议直接使用实体类)
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnv.toRetractStream(tableResult, Row.class);
        tuple2DataStream.print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
